package com.glodon.paas.document.storage.directory;

import static org.apache.commons.io.IOUtils.*;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.aliyun.openservices.ClientException;
import com.aliyun.openservices.oss.OSSClient;
import com.aliyun.openservices.oss.OSSException;
import com.aliyun.openservices.oss.model.AbortMultipartUploadRequest;
import com.aliyun.openservices.oss.model.CompleteMultipartUploadRequest;
import com.aliyun.openservices.oss.model.GetObjectRequest;
import com.aliyun.openservices.oss.model.InitiateMultipartUploadRequest;
import com.aliyun.openservices.oss.model.InitiateMultipartUploadResult;
import com.aliyun.openservices.oss.model.ListPartsRequest;
import com.aliyun.openservices.oss.model.OSSObject;
import com.aliyun.openservices.oss.model.ObjectMetadata;
import com.aliyun.openservices.oss.model.PartETag;
import com.aliyun.openservices.oss.model.PartListing;
import com.aliyun.openservices.oss.model.PartSummary;
import com.aliyun.openservices.oss.model.UploadPartRequest;
import com.glodon.paas.document.storage.id.ObjectResult;
import com.glodon.paas.document.storage.util.DigestState;
import com.glodon.paas.document.storage.util.StorageUtil;

public class OSSDirectory extends AbstractDirectory {
    private static final Logger logger = LoggerFactory.getLogger(OSSDirectory.class);;

    private String bucketName;
    private File tempDir;
    private OSSClient ossClient;

    private String endpoint;
    private String accessId;
    private String accessKey;

    public OSSDirectory() {
    }

    public OSSDirectory(String endpoint, String accessId, String accessKey, String bucketName) throws IOException {
        this.bucketName = bucketName;
        this.endpoint = endpoint;
        this.accessId = accessId;
        this.accessKey = accessKey;
        init();
    }

    public void init() {
        tempDir = new File(System.getProperty("java.io.tmpdir"));
        ossClient = new OSSClient(endpoint, accessId, accessKey);
    }

    @Override
    public ObjectResult save(InputStream is) throws IOException {
        FileInputStream tempIn = null;
        try {
            File tempFile = createTempFile();
            FileOutputStream out = new FileOutputStream(tempFile);
            ObjectResult objectResult = StorageUtil.toOut(is, out);
            String objectId = objectResult.getObjectId();
            out.close();
            if (has(objectId)) {
                tempFile.delete();
            } else {
                String key = getKey(objectId);
                ObjectMetadata objectMeta = new ObjectMetadata();
                objectMeta.setContentLength(tempFile.length());
                tempIn = new FileInputStream(tempFile);
                getAudit().addObject(key);
                ossClient.putObject(bucketName, key, tempIn, objectMeta);
            }
            return objectResult;
        } catch (OSSException e) {
            throw new IOException(e.getMessage(), e);
        } catch (ClientException e) {
            throw new IOException(e.getMessage(), e);
        } finally {
            IOUtils.closeQuietly(tempIn);
        }
    }

    private File createTempFile() throws IOException {
        File nozDir = new File(tempDir, NOZ);
        FileUtils.forceMkdir(nozDir);
        return File.createTempFile(NOZ, "", nozDir);
    }

    @Override
    public boolean has(String id) throws IOException {
        try {
            ObjectMetadata objectMetadata = getObjectMetadata(id);
            return objectMetadata != null;
        } catch (ClientException e) {
            throw new IOException(e.getMessage(), e);
        }
    }

    private ObjectMetadata getObjectMetadata(String objectId) throws ClientException {
        try {
            return ossClient.getObjectMetadata(bucketName, getKey(objectId));
        } catch (OSSException e) {
            return null;
        }
    }

    @Override
    public InputStream get(String id) throws IOException {
        try {
            if (!has(id))
                return null;
            GetObjectRequest request = new GetObjectRequest(bucketName, getKey(id));
            OSSObject result = ossClient.getObject(request);
            InputStream oIn = result.getObjectContent();
            return StorageUtil.getStorageInputStream(oIn);
        } catch (OSSException e) {
            throw new IOException(e.getMessage(), e);
        } catch (ClientException e) {
            throw new IOException(e.getMessage(), e);
        }
    }

    @Override
    public InputStream get(String id, long start) throws IOException {
        try {
            if (!has(id))
                return null;
            ObjectMetadata md = ossClient.getObjectMetadata(bucketName, getKey(id));
            long length = md.getContentLength();
            GetObjectRequest request = new GetObjectRequest(bucketName, getKey(id));
            request.setRange(start, length - 1);
            OSSObject result = ossClient.getObject(request);
            InputStream oIn = result.getObjectContent();
            return StorageUtil.getStorageInputStream(oIn);
        } catch (OSSException e) {
            throw new IOException(e.getMessage(), e);
        } catch (ClientException e) {
            throw new IOException(e.getMessage(), e);
        }
    }

    @Override
    public String initPart() throws IOException {
        try {
            String uuid = StorageUtil.generateUUID();
            String path = getPartPath(uuid);
            InitiateMultipartUploadRequest initiatemultipartuploadrequest = new InitiateMultipartUploadRequest(
                    bucketName, path);
            InitiateMultipartUploadResult result = ossClient.initiateMultipartUpload(initiatemultipartuploadrequest);
            return getPartId(uuid, result);
        } catch (OSSException e) {
            throw new IOException(e.getMessage(), e);
        } catch (ClientException e) {
            throw new IOException(e.getMessage(), e);
        }
    }

    private String getPartId(String uuid, InitiateMultipartUploadResult result) {
        return uuid + "_" + result.getUploadId();
    }

    @Override
    public ObjectResult append(String partId, int partNumber, String state, InputStream is, long length)
            throws IOException {
        logger.debug("append part: part id = {}, state = {}, length = {}", partId, state, length);
        File tempFile = createTempFile();
        try {
            String[] partIds = StringUtils.split(partId, '_');
            String uuid = partIds[0];
            String uploadId = partIds[1];
            String path = getPartPath(uuid);
            FileOutputStream out = new FileOutputStream(tempFile);
            DigestState digestState;
            if (state != null) {
                digestState = DigestState.getDigestState(state);
            } else {
                digestState = new DigestState();
            }
            MessageDigest md = digestState.getMessageDigest();
            ObjectResult objectResult = StorageUtil.toOut(is, out, md);
            closeQuietly(out);
            long l = objectResult.getLength();
            if (l < length) {
                throw new IOException("append file error, expect length: " + length + ", actual: " + l);
            }
            List<PartSummary> parts = getParts(path, uploadId);
            if (parts.size() != partNumber - 1)
                throw new IOException("part number is invalid: part size = " + parts.size() + ", part number = "
                        + partNumber);
            logger.debug("append part: part id = {}, state = {}, length = {}, part size = {}", partId, state, length,
                    parts.size());
            UploadPartRequest uploadPartRequest = new UploadPartRequest();
            uploadPartRequest.setBucketName(bucketName);
            uploadPartRequest.setKey(path);
            uploadPartRequest.setUploadId(uploadId);
            uploadPartRequest.setPartNumber(parts.size() + 1);
            uploadPartRequest.setPartSize(objectResult.getLength());
            FileInputStream in = new FileInputStream(tempFile);
            uploadPartRequest.setInputStream(in);
            ossClient.uploadPart(uploadPartRequest);
            IOUtils.closeQuietly(in);
            logger.debug("append part, upload oss success: part id = {}, state = {}, length = {}, part size = {}",
                    partId, state, length, parts.size());
            return objectResult;
        } catch (OSSException e) {
            throw new IOException(e.getMessage(), e);
        } catch (ClientException e) {
            throw new IOException(e.getMessage(), e);
        } finally {
            tempFile.delete();
        }
    }

    private List<PartSummary> getParts(String path, String id) {
        try {
            ListPartsRequest request = new ListPartsRequest(bucketName, path, id);
            PartListing partList = ossClient.listParts(request);
            List<PartSummary> parts = partList.getParts();
            return parts;
        } catch (Exception e) {
            return new ArrayList<PartSummary>();
        }
    }

    @Override
    public void complete(String partId, String objectId) throws IOException {
        logger.debug("complete part: part id = {}, object id = {}", partId, objectId);
        try {
            String[] partIds = StringUtils.split(partId, '_');
            String uuid = partIds[0];
            String uploadId = partIds[1];
            String path = getPartPath(uuid);
            ListPartsRequest request = new ListPartsRequest(bucketName, path, uploadId);
            PartListing partList = ossClient.listParts(request);
            ArrayList<PartETag> etags = new ArrayList<PartETag>();
            List<PartSummary> parts = partList.getParts();
            for (PartSummary partSummary : parts) {
                etags.add(new PartETag(partSummary.getPartNumber(), partSummary.getETag()));
            }
            CompleteMultipartUploadRequest completeMultipartUploadRequest = new CompleteMultipartUploadRequest(
                    bucketName, path, uploadId, etags);
            ossClient.completeMultipartUpload(completeMultipartUploadRequest);
            logger.debug("complete part, complete oss success: part id = {}, object id = {}", partId, objectId);
            if (!has(objectId)) {
                String key = getKey(objectId);
                getAudit().addObject(key);
                ossClient.copyObject(bucketName, path, bucketName, key);
                logger.debug("complete part, copy oss success: part id = {}, object id = {}", partId, objectId);
            }
            ossClient.deleteObject(bucketName, path);
            logger.debug("complete part, delete oss success: part id = {}, object id = {}", partId, objectId);
        } catch (OSSException e) {
            throw new IOException(e.getMessage(), e);
        } catch (ClientException e) {
            throw new IOException(e.getMessage(), e);
        }
    }

    @Override
    public void abortPart(String partId) throws IOException {
        try {
            String[] partIds = StringUtils.split(partId, '_');
            String uuid = partIds[0];
            String uploadId = partIds[1];
            String path = getPartPath(uuid);
            AbortMultipartUploadRequest request = new AbortMultipartUploadRequest(bucketName, path, uploadId);
            ossClient.abortMultipartUpload(request);
        } catch (OSSException e) {
            throw new IOException(e.getMessage(), e);
        } catch (ClientException e) {
            throw new IOException(e.getMessage(), e);
        }
    }

    public void setBucketName(String bucketName) {
        this.bucketName = bucketName;
    }

    public void setEndpoint(String endpoint) {
        this.endpoint = endpoint;
    }

    public void setAccessId(String accessId) {
        this.accessId = accessId;
    }

    public void setAccessKey(String accessKey) {
        this.accessKey = accessKey;
    }
}
